package com.jerome.fs.server.aggregation;

import com.jerome.fs.config.ConfigUtil;
import com.jerome.fs.http.HttpMethodEnum;
import com.jerome.fs.http.server.HttpServer;
import com.jerome.fs.json.JsonParseException;
import com.jerome.fs.server.aggregation.constants.AggregationConstant;
import com.jerome.fs.server.aggregation.handler.AggregationHttpHandler;
import com.jerome.fs.server.aggregation.handler.AggregationMqServer;
import com.jerome.fs.server.aggregation.handler.SlaveSyncClientHandler;
import com.jerome.fs.sync.SyncClient;
import com.jerome.fs.util.ThreadPoolFactory;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Aggregation Server
 * It exposes two interfaces
 * GET /weather/{id}
 * PUT /weather
 *
 * @author Jerome
 * @date 2023/7/31
 */
public class AggregationServer {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InstantiationException, IllegalAccessException, JsonParseException {
        Map<String, String> paramsMap = ConfigUtil.getMap(args);
        String config_path = paramsMap.get(AggregationConstant.CONFIG_PATH);
        if (config_path == null) {
            System.out.println("Pass the parameter config_path, such as config_path=/tmp/config.properties");
        }
        // Initialize configuration information
        ContextUtil.init(config_path);
        if (ContextUtil.getAggregationConfig().getMode().equals(AggregationConstant.MASTER)){
            // Start the master node
            startMaster();
        } else {
            // Start the slave node
            startSlave();
        }
    }

    /**
     * Start the master node
     */
    private static void startMaster() throws ClassNotFoundException, InstantiationException, IllegalAccessException, JsonParseException, IOException {
        // Recover data
        ContextUtil.restoreDBF();
        // Start the HTTP server
        // It exposes two interfaces, GET /weather/{id} and PUT /weather
        HttpServer.startHttpServer(ContextUtil.getAggregationConfig().getPort(), new AggregationHttpHandler());
        // Start MQ Consumer Service
        // Use MQ control to accept requests for processing in Lamport clock order
        new AggregationMqServer().start();
        // Enable DBF to back up data in a timely manner
        ExecutorService taskExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1),
                new ThreadPoolFactory("AggregationServer_dbf"),
                new ThreadPoolExecutor.DiscardPolicy());
        taskExecutor.submit(new DBFThread());
        taskExecutor.shutdown();
    }

    /**
     * Start the slave node
     * The slave node mainly synchronizes data with the master node and accepts GET requests
     * @throws IOException
     */
    private static void startSlave() throws IOException, ClassNotFoundException {
        // 1. Register heartbeat detection
        SlaveSyncClientHandler slaveSyncClientHandler = new SlaveSyncClientHandler();
        SyncClient client = new SyncClient(ContextUtil.getAggregationConfig().getMasterAddress(), slaveSyncClientHandler, HttpMethodEnum.REGISTER);
        // Start the HTTP server
        // It exposes two interfaces, GET /weather/{id} and PUT /weather
        HttpServer.startHttpServer(ContextUtil.getAggregationConfig().getPort(), new AggregationHttpHandler());
        new AggregationMqServer().start();
        // 2. Full synchronization
        client.syncAll();
        // 3. Enable incremental synchronization
        int i = 0;
        while (true){
            try {
                client.incrementSync(0);
                Thread.sleep(5);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

    /**
     * The main thread task processing class
     * DBF  data base to file
     * Write cached data to a file
     *
     * @author Jerome
     * @version 1.0
     * @date 2023/9/8 11:02
     */
    private static class DBFThread implements Runnable {
        public DBFThread() {
        }

        @Override
        public void run() {
            // Execute DBF to back up in-memory data to a file
            while (true) {
                try {
                    ContextUtil.serializeDBF();
                    // Back up every 5 seconds
                    Thread.sleep(5000);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }
}
